package com.aotain.kafka;

import java.util.ArrayList;
import java.util.List;

import org.apache.log4j.Logger;
import org.apache.storm.kafka.SpoutConfig;
import org.apache.storm.kafka.StringScheme;
import org.apache.storm.kafka.ZkHosts;
import org.apache.storm.spout.SchemeAsMultiScheme;

public class KafkaConfigUtil {

	String topic = null;
	static SpoutConfig spoutConfig = null;
	public KafkaConfigUtil(String topic,
			String zkRoot,String id) {
		this.topic = topic;
		ZkHosts zkHosts = new ZkHosts(KafkaProperties.zkConnect);
		zkHosts.brokerZkPath = "/kafka_2/brokers";
		spoutConfig = new SpoutConfig(zkHosts,topic,zkRoot,id);
		List<String> zkServers = new ArrayList<String>() ;
		for(String host : zkHosts.brokerZkStr.split(","))
		{
			zkServers.add(host.split(":")[0]);
		}
		spoutConfig.zkServers = zkServers ;
		spoutConfig.zkPort = 2181;
		spoutConfig.ignoreZkOffsets = false; //
		spoutConfig.socketTimeoutMs = 60 * 1000 ;//
		spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme()) ; //

		//spoutConfig.TOPOLOGY_ACKERS
		//spoutConfig.topic = this.topic;
		
	}
	
	public KafkaConfigUtil(String zkServer,String topic,
			String zkRoot,String id) {
		this.topic = topic;
		ZkHosts zkHosts = new ZkHosts(zkServer);
		zkHosts.brokerZkPath = "/kafka_2/brokers";
		spoutConfig = new SpoutConfig(zkHosts,topic,zkRoot,id);
		List<String> zkServers = new ArrayList<String>() ;
		for(String host : zkHosts.brokerZkStr.split(","))
		{
//			zkServers.add(host.split(":")[0]);
			zkServers.add(host);
			Logger.getLogger(KafkaConfigUtil.class).info("brokerZk:" + host);
		}
		spoutConfig.zkServers = zkServers ;
		spoutConfig.zkPort = 2181;
		spoutConfig.ignoreZkOffsets = false; //
		spoutConfig.socketTimeoutMs = 60 * 1000 ;//
		spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme()) ; //
		spoutConfig.startOffsetTime = kafka.api.OffsetRequest.LatestTime();//从何时的offset时间开始读，默认为最旧的offset(Ealiest)
		//spoutConfig.maxOffsetBehind = 50000;
		//spoutConfig.topic = this.topic;
		
	}
	
	
	
	public SpoutConfig getConfig() {
		return spoutConfig;
	}
}
